package Service

import Config.Constant
import Entity.Model.Anime
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.json4s._
import org.json4s.native.JsonMethods._
import redis.clients.jedis.Jedis

import java.time.{Duration, LocalDateTime}
import java.time.format.DateTimeFormatter
import java.util.Properties
object Project_Day_total {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val proper = new Properties()
    proper.setProperty("bootstrap.servers", Constant.KAFKA_SERVER_BIND)
    proper.setProperty("group.id",Constant.KAFKA_GROUP_ID)
    val data=env
      .addSource(new FlinkKafkaConsumer[String](Constant.KAFKA_TOPIC,new SimpleStringSchema(),proper))
      .map(data=>{
        implicit val formats=DefaultFormats
        val parseJson=parse(data,useBigDecimalForDouble = true)
        parseJson.extract[Anime]
      })
      .assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness[Anime](Duration.ofSeconds(1))
        .withTimestampAssigner(
          new SerializableTimestampAssigner[Anime] {
            override def extractTimestamp(t: Anime, l: Long): Long = LocalDateTime.parse(t.Date, DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")).toInstant(java.time.ZoneOffset.UTC).toEpochMilli
          }
        ))
      .map(data=>{("anime",1)})
      .keyBy(_._1)
      .window(SlidingEventTimeWindows.of(Time.days(1),Time.seconds(1)))
      .reduce((r1,r2)=>{(r1._1,r1._2+r2._2)})
      .map(data=>{(data._1,data._2.toLong)})
      .addSink(new UserRichSinkFunction)
    env.execute()
  }
  class UserRichSinkFunction extends RichSinkFunction[(String,Long)] {
    var jedis:Jedis=_
    override def open(parameters: Configuration): Unit = {
      jedis=new Jedis(Constant.REDIS_BIND,Constant.REDIS_POST)
      jedis.select(Constant.REDIS_DATABASE)
    }
    override def invoke(in:(String,Long)): Unit = {
      jedis.publish(Constant.Project_Day_total,in._2.toString)
      jedis.set(Constant.Project_Day_total,in._2.toString)
      println(in)
    }

    override def close(): Unit = {
      jedis.close()
    }
  }
}
